package com.atguigu.hbase.api;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * Hbase Api
 *
 * 1、创建客户端对象
 * 2、调用api方法
 * 3、关闭资源
 *
 */
public class HbaseApiDemo {

    public static Connection connection = null ;

    static {
        try {
            //new Configuration();
            Configuration configuration = HBaseConfiguration.create();
            //客户端通过zk找到hbase
            configuration.set( "hbase.zookeeper.quorum" , "hadoop102,hadoop103,hadoop104");
            connection = ConnectionFactory.createConnection(configuration) ;
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws IOException {
        //System.out.println(connection);
        //testCreateTable(null , "emp" , "f1" , "f2");

        //testPutData(null , "emp" , "1001" , "f1" , "name" , "zhangxiaosan");
        //testDeleteData(null , "emp" , "1001" , "f1" , "name");

        //testGetData(null , "stu" , "1001");
        testScanData(null , "stu" , "10021" , "1003!");

        connection.close();
    }
    /**
     * DML : 查询  get / scan
     *
     * shell : get 'namespaceName:tableName' , 'rk'
     *
     *         scan 'namespaceName:tableName'  , {STARTROW=> '' , STOPROW => ''}
     *
     *
     * 补充: 其实scan操作可以通过Filter的方式写各种各样的过滤操作，但是一般不建议用， 因为除了rk之外的过滤操作全部都是全表扫描， 性能太差。
     *
     */
    public static void testScanData(String namespaceName, String tableName , String startRow , String stopRow ) throws IOException {
        // 获取Table对象
        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = connection.getTable(tn);
        Scan scan = new Scan();
        //scan.withStartRow(Bytes.toBytes(startRow)).withStopRow(Bytes.toBytes(stopRow));
        ResultScanner scanner = table.getScanner(scan);
        //通过迭代的方式处理每条数据
        for (Result result : scanner) {
            //每条数据中的一个列封装一个Cell对象
            List<Cell> cells = result.listCells();
            for (Cell cell : cells) {
                String kv = Bytes.toString(CellUtil.cloneRow(cell)) + " : " +
                        Bytes.toString(CellUtil.cloneFamily(cell)) + " : " +
                        Bytes.toString(CellUtil.cloneQualifier(cell)) + " : " +
                        Bytes.toString(CellUtil.cloneValue(cell));

                System.out.println(kv);
            }

            System.out.println("---------------------------------");
        }

        table.close();
    }



    public static void testGetData(String namespaceName, String tableName , String rk ) throws IOException {
        // 获取Table对象
        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = connection.getTable(tn);
        Get get = new Get(Bytes.toBytes(rk));
        //指定查询具体的列族 或者 列
        // get.addFamily();
        // get.addColumn();

        //每条数据封装一个Result对象
        Result result = table.get(get);
        //每条数据中的一个列封装一个Cell对象
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            String kv = Bytes.toString(CellUtil.cloneRow(cell)) + " : " +
                    Bytes.toString(CellUtil.cloneFamily(cell)) + " : " +
                    Bytes.toString(CellUtil.cloneQualifier(cell)) + " : " +
                    Bytes.toString(CellUtil.cloneValue(cell));

            System.out.println(kv);
        }

        table.close();
    }




    /**
     * DML : 删除
     *
     * shell :
     *      delete    'namespaceName:tableName' , 'rk' , 'cf:cl'  => Delete
     *      deleteall 'namespaceName:tableName' , 'rk' , 'cf:cl'  => DeleteColumn / DeleteFamily
     */
    public static void testDeleteData(String namespaceName, String tableName , String rk , String cf, String cl ) throws IOException {
        // 获取Table对象
        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = connection.getTable(tn);
        Delete delete = new Delete(Bytes.toBytes(rk));
        //Delete
        //delete.addColumn(Bytes.toBytes(cf) , Bytes.toBytes(cl));

        //DeleteColumn
        //delete.addColumns(Bytes.toBytes(cf) , Bytes.toBytes(cl));

        //DeleteFamily
        //delete.addFamily("f2")
        table.delete(delete);


        table.close();
    }



    /**
     * DML : 新增、修改
     *
     * shell : put 'namespaceName:tableName' , 'rk' , 'cf:cl' , 'v'
     */
    public static void  testPutData(String namespaceName, String tableName , String rk ,String cf , String cl , String v ) throws IOException {
        // 获取Table对象
        TableName tn = TableName.valueOf(namespaceName, tableName);
        Table table = connection.getTable(tn);

        Put put = new Put(Bytes.toBytes(rk));
        //设置列族 列 值
        put.addColumn(Bytes.toBytes(cf) , Bytes.toBytes(cl) , Bytes.toBytes(v));
        table.put(put);

        table.close();
    }




    /**
     * DDL :  建表
     *
     * shell : create 'namespaceName:tableName' , 'cf' ...
     *
     * 自己尝试完成:
     * 创建库  删除库
     * 创建表  删除表  修改表 ....
     *
     */
    public static void testCreateTable(String namespaceName , String tableName , String ... cfs ) throws IOException {
        //基础的判空操作
        if(tableName == null || tableName.equals("")){
            throw new RuntimeException("表名不能为空");
        }
        if(cfs == null || cfs.length <= 0){
            throw  new RuntimeException("至少指定一个列族");
        }
        //1. 获取Admin对象
        Admin admin = connection.getAdmin();

        // 判断表是否存在
        TableName tn = TableName.valueOf(namespaceName, tableName);  // 如果namespace的名字为null， 自动处理成default
        boolean exists = admin.tableExists(tn);
        if(exists){
            throw new RuntimeException( ((namespaceName == null || namespaceName.equals("")) ?"default":namespaceName) + ":" + tableName + " 已经存在!!!");
        }

        // 创建表
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tn);
        //设置列族
        for (String cf : cfs) {
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
            ColumnFamilyDescriptor familyDescriptor = columnFamilyDescriptorBuilder.build();
            tableDescriptorBuilder.setColumnFamily(familyDescriptor);
        }

        TableDescriptor tableDescriptor = tableDescriptorBuilder.build();

        admin.createTable(tableDescriptor);

        System.out.println(((namespaceName == null || namespaceName.equals("")) ?"default":namespaceName) + ":" + tableName + " 创建成功!!!");

        admin.close();

    }

}
